package com.demo.quickstart;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class StreamingJob {
    @SuppressWarnings("lambda")
    public static void main(String[] args) throws Exception {
        // 创建Flink的流式计算环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 监听本地9000端口
        DataStream<String> text = env.socketTextStream("127.0.0.1", 9000, "\n");
        // 将接收的数据进行拆分，分组，窗口计算并且进行聚合输出; flatMap中的参数这里不能使用lambda表达式
        DataStream<WordWithCount> windowCounts = text
                .flatMap(new SplitFlatMapFunction())
                .keyBy("word")
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .reduce((ReduceFunction<WordWithCount>) (a, b) -> new WordWithCount(a.word, a.count + b.count));
        // 打印结果
        windowCounts.print().setParallelism(1);
        env.execute("Socket Window WordCount");
    }


    // Data type for words with count
    public static class WordWithCount {
        public String word;
        public long count;

        @SuppressWarnings("unused")
        // 此无参构造方法得有，不然启动会报错
        public WordWithCount() {
        }

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + " : " + count;
        }
    }

    public static class SplitFlatMapFunction implements FlatMapFunction<String, WordWithCount> {
        @Override
        public void flatMap(String value, Collector<WordWithCount> collector) {
            String[] split = value.split("\\s");
            for (String word : split) {
                collector.collect(new WordWithCount(word, 1L));
            }
        }
    }

}
